package com.hhf.rpc.core.registry.impl;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.collection.ConcurrentHashSet;
import cn.hutool.core.util.StrUtil;
import cn.hutool.cron.CronUtil;
import cn.hutool.cron.task.Task;
import cn.hutool.json.JSONUtil;
import com.hhf.rpc.core.config.RegistryConfig;
import com.hhf.rpc.core.model.ServiceMetaInfo;
import com.hhf.rpc.core.registry.Registry;
import com.hhf.rpc.core.registry.RegistryServiceCache;
import io.etcd.jetcd.*;
import io.etcd.jetcd.options.GetOption;
import io.etcd.jetcd.options.PutOption;
import io.etcd.jetcd.watch.WatchEvent;
import lombok.extern.slf4j.Slf4j;

import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * @Author hhf
 * @Date 2024/10/17
 * @Description etcd注册中心
 */
@Slf4j
public class EtcdRedistry implements Registry {

    private Client client;

    private KV kvClient;

    /**
     * 根节点
     */
    private static final String ETCD_ROOT_PATH = "/rpc/";

    /**
     * 本地服务缓存
     */
    private final RegistryServiceCache registryServiceCache = new RegistryServiceCache();

    /**
     * 本机注册节点的key,用于维护续期
     */
    private final Set<String> localRegisterNodeKeySet = new HashSet<>();

    /**
     * 正在监听的key集合
     */
    private final Set<String> watchingKeySet = new ConcurrentHashSet<>();

    @Override
    public void init(RegistryConfig registryConfig) {
        String address = registryConfig.getAddress();
        Long timeout = registryConfig.getTimeout();
        client = Client.builder().endpoints(address).connectTimeout(Duration.ofMillis(timeout)).build();
        kvClient = client.getKVClient();

        //开启定时任务(心跳检测)
        heartBeat();
    }

    @Override
    public void register(ServiceMetaInfo serviceMetaInfo) throws Exception {
        //1.创建 Least 客户端
        Lease leaseClient = client.getLeaseClient();
        //2.创建 30s 的租约
        long leaseId = leaseClient.grant(30).get().getID();
        //3.设置要存储的键值对
        String serviceKey = serviceMetaInfo.getServiceNodeKey();
        String registerKey = ETCD_ROOT_PATH + serviceKey;
        ByteSequence key = ByteSequence.from(registerKey, StandardCharsets.UTF_8);
        ByteSequence value = ByteSequence.from(JSONUtil.toJsonStr(serviceMetaInfo), StandardCharsets.UTF_8);
        //4.将键值对与租约关联起来=>设置过期时间
        PutOption putOption = PutOption.builder().withLeaseId(leaseId).build();
        kvClient.put(key, value, putOption).get();
        //5.添加节点信息至本地缓存
        localRegisterNodeKeySet.add(registerKey);
    }

    @Override
    public void unRegister(ServiceMetaInfo serviceMetaInfo) throws Exception {
        //1.获取 key
        String serviceKey = serviceMetaInfo.getServiceNodeKey();
        String registerKey = ETCD_ROOT_PATH + serviceKey;
        ByteSequence key = ByteSequence.from(registerKey, StandardCharsets.UTF_8);
        //2.从etcd中删除
        kvClient.delete(key).get();
        //3.从本地缓存中删除
        localRegisterNodeKeySet.remove(registerKey);
    }

    @Override
    public List<ServiceMetaInfo> serviceDiscovery(String serviceKey) throws Exception {

        //优先从缓存中查找服务
        List<ServiceMetaInfo> serviceMetaInfos = registryServiceCache.readCache(serviceKey);
        if (!CollUtil.isEmpty(serviceMetaInfos)) {
            return serviceMetaInfos;
        }

        //1.获取 服务查询前缀
        String searchPrefix = ETCD_ROOT_PATH + serviceKey + StrUtil.C_SLASH;
        //2.根据前缀查询
        GetOption getOption = GetOption.builder().isPrefix(true).build();
        List<KeyValue> keyValues = kvClient.get(ByteSequence.from(searchPrefix, StandardCharsets.UTF_8), getOption)
                .get()
                .getKvs();
        //3.解析
        List<ServiceMetaInfo> serviceMetaInfoList = keyValues.stream()
                .map(keyValue -> {

                    String key = keyValue.getKey().toString(StandardCharsets.UTF_8);
                    //监听key
                    watch(key);

                    String value = keyValue.getValue().toString(StandardCharsets.UTF_8);
                    return JSONUtil.toBean(value, ServiceMetaInfo.class);
                })
                .collect(Collectors.toList());
        //4.判断
        if (CollectionUtil.isEmpty(serviceMetaInfoList)) {
            throw new RuntimeException("未发现任何服务");
        }

        //写入缓存服务
        registryServiceCache.writeCache(serviceKey,serviceMetaInfoList);

        //5.返回结果
        return serviceMetaInfoList;
    }

    @Override
    public void destory() {
        //下线所有节点
        for (String key : localRegisterNodeKeySet) {
            try {
                kvClient.delete(ByteSequence.from(key, StandardCharsets.UTF_8)).get();
            } catch (Exception e) {
                e.printStackTrace();
                log.error("节点{}下线失败", key);
            }
        }

        log.info("当前{}结点下线", client.toString());
        if (client != null) {
            client.close();
        }
        if (kvClient != null) {
            kvClient.close();
        }
    }

    @Override
    public void heartBeat() {
        //10s 续签一次
        CronUtil.schedule("*/10 * * * * *", new Task() {
            @Override
            public void execute() {
                //遍历本节点的所有key
                for (String key : localRegisterNodeKeySet) {
                    try {
                        List<KeyValue> keyValues = kvClient.get(ByteSequence.from(key, StandardCharsets.UTF_8))
                                .get()
                                .getKvs();
                        //如果节点过期,continue
                        if (CollUtil.isEmpty(keyValues)) {
                            continue;
                        }
                        //未过期,续期
                        KeyValue keyValue = keyValues.get(0);
                        String value = keyValue.getValue().toString(StandardCharsets.UTF_8);
                        ServiceMetaInfo serviceMetaInfo = JSONUtil.toBean(value, ServiceMetaInfo.class);
                        register(serviceMetaInfo);

                    } catch (Exception e) {
                        throw new RuntimeException(key + "续签失败", e);
                    }
                }
            }
        });


        //开启秒级别定时任务
        CronUtil.setMatchSecond(true);
        //启动定时任务
        CronUtil.start();
    }

    @Override
    public void watch(String serviceNodeKey) {

        Watch watchClient = client.getWatchClient();
        //之前未被监听,则开启监听
        boolean newWatch = watchingKeySet.add(serviceNodeKey);
        if (newWatch) {
            watchClient.watch(ByteSequence.from(serviceNodeKey, StandardCharsets.UTF_8), response -> {
                for (WatchEvent event : response.getEvents()) {
                    switch (event.getEventType()) {
                        case DELETE -> {
                            //清空注册服务
                            registryServiceCache.clearCache(serviceNodeKey);
                            break;
                        }
                        default -> {
                            break;
                        }
                    }
                }
            });
        }

    }
}
